package com.upcoming.data.storm.caller;

import org.apache.storm.Config;
import org.apache.storm.StormSubmitter;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

import java.util.HashMap;
import java.util.Map;

/**
 * 作者：tianjiayuan
 * 创建时间：2018-05-17 13:44
 * 类描述：
 * 修改人：
 * 修改时间：
 */
//main  com.upcoming.data.storm.caller.LogAnalyserStorm
public class LogAnalyserStorm {

    public static void main(String[] args) throws Exception{

        //配置信息
        Config conf = new Config();
        conf.setDebug(false);
        Map<String, String> map = new HashMap<>();
        map.put("storm.zookeeper.servers", "127.0.0.1");
        conf.setEnvironment(map);

        TopologyBuilder builder = new TopologyBuilder();
        //设置水龙头
        builder.setSpout("call-log-reader-spout", new FakeCallLogReaderSpout(), 1);
        //设置转换器并重洗分组 随机分发
        builder.setBolt("call-log-creator-bolt", new CallerCreatorBolt(), 3).shuffleGrouping("call-log-reader-spout");
        //字段分组，字段值相同分发到同一个bolt
//        builder.setBolt("call-log-creator-bolt", new CallerCreatorBolt(), 3).fieldsGrouping("call-log-reader-spout", new Fields("from"));
        //每个bolt都会收到bolt
//        builder.setBolt("call-log-creator-bolt", new CallerCreatorBolt(), 3).allGrouping("call-log-reader-spout");
//        builder.setBolt("call-log-creator-bolt", new CallerCreatorBolt(), 3).customGrouping("call-log-reader-spout", new MyGrouping());
        //设置bolt并字段分组
        //设置分组策略
        builder.setBolt("call-log-count-bolt", new CallLogCounterBolt(), 3).fieldsGrouping("call-log-creator-bolt",
                new Fields("call"));

        //运行本地集群模式
        /*LocalCluster localCluster = new LocalCluster();
        //向集群中提交Topology
        localCluster.submitTopology("LogAnalyserStorm", conf, builder.createTopology());
        Thread.sleep(1000*60);
        localCluster.shutdown();*/

        //线上提交
        StormSubmitter.submitTopology("APPStorm", conf, builder.createTopology());
    }
}
